package org.zhaya.sdk.processors;

import cn.hutool.core.util.StrUtil;
import com.google.common.base.MoreObjects;
import com.google.common.primitives.Doubles;
import com.google.common.primitives.Longs;
import io.logz.sawmill.Doc;
import io.logz.sawmill.ProcessResult;
import io.logz.sawmill.Processor;
import io.logz.sawmill.annotations.ProcessorProvider;
import io.logz.sawmill.exceptions.ProcessorConfigurationException;
import io.logz.sawmill.utilities.JsonUtils;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.zhaya.sdk.util.JsonProcessorsUtils;

import java.util.*;

import static java.util.Objects.requireNonNull;

@ProcessorProvider(type = "convert", factory = ConvertFieldProcessor.Factory.class)
public class ConvertFieldProcessor implements Processor {

    private final List<String> paths;
    private final FieldType fieldType;

    public ConvertFieldProcessor(List<String> paths, FieldType fieldType) {
        this.paths = requireNonNull(paths, "paths cannot be null");
        this.fieldType = requireNonNull(fieldType, "field type cannot be null");
    }

    public FieldType getFieldType() {
        return fieldType;
    }

    @Override
    public ProcessResult process(Doc doc) {
        List<String> errorMessages = new ArrayList<>();
        for (String path : paths) {
            if (!doc.hasField(path)) {
                errorMessages.add(String.format("failed to convert field in path [%s], field is missing.", path));
                continue;
            }
            Object beforeCast = doc.getField(path);

            Object afterCast;
            if (fieldType == FieldType.BOOLEAN || fieldType == FieldType.STRING || fieldType == FieldType.JSONSTRING) {
                afterCast = fieldType.convertFrom(beforeCast);
            } else {
                afterCast = fieldType.convertFrom(beforeCast, 0L);
            }

            if (Objects.isNull(afterCast)) {
                errorMessages.add(String.format("failed to convert field in path [%s] to %s, value [%s].", path, fieldType, beforeCast));
                continue;
            }

            boolean succeeded = doc.removeField(path);
            if (succeeded) {
                doc.addField(path, afterCast);
            } else {
                errorMessages.add(String.format("failed to convert field in path [%s] to %s, value [%s].", path, fieldType, beforeCast));
            }
        }
        if (errorMessages.isEmpty()) {
            return ProcessResult.success();
        }
        String allErrorMessages = String.join(StrUtil.LF, errorMessages);
        return ProcessResult.failure(allErrorMessages);
    }

    public static class Factory implements Processor.Factory {

        @Override
        public ConvertFieldProcessor create(Map<String, Object> config) {

            Configuration convertFieldConfig = JsonUtils.fromJsonMap(Configuration.class, config);

            if (ObjectUtils.allNull(convertFieldConfig.getPath(), convertFieldConfig.getPaths())) {
                throw new ProcessorConfigurationException("failed to parse convert processor config, could not resolve field path/s");
            }

            if (ObjectUtils.allNotNull(convertFieldConfig.getPath(), convertFieldConfig.getPaths())) {
                throw new ProcessorConfigurationException("failed to parse convert processor config, both field path and field paths are defined when only 1 allowed");
            }

            if (Objects.isNull(convertFieldConfig.getType())) {
                throw new ProcessorConfigurationException("failed to parse convert processor config, could not resolve field type");
            }

            List<String> paths = Objects.isNull(convertFieldConfig.getPath()) ? convertFieldConfig.getPaths() : Collections.singletonList(convertFieldConfig.getPath());

            return new ConvertFieldProcessor(paths, convertFieldConfig.getType());
        }
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    @Builder
    public static class Configuration implements Processor.Configuration {
        private String path;
        private List<String> paths;
        private FieldType type;

    }

    public enum FieldType {
        INT {
            @Override
            public Object convertFrom(Object value) {
                return Longs.tryParse(String.valueOf(value));
            }
        },
        LONG {
            @Override
            public Object convertFrom(Object value) {
                return Longs.tryParse(String.valueOf(value));
            }
        },
        FLOAT {
            @Override
            public Object convertFrom(Object value) {
                return Doubles.tryParse(String.valueOf(value));
            }
        },
        DOUBLE {
            @Override
            public Object convertFrom(Object value) {
                return Doubles.tryParse(String.valueOf(value));
            }
        },
        STRING {
            @Override
            public Object convertFrom(Object value) {
                return String.valueOf(value);
            }
        },
        JSONSTRING {
            @Override
            public Object convertFrom(Object value) {
                if (value instanceof String) {
                    String jsonString = JsonProcessorsUtils.toJsonString(value);
                    if (StringUtils.isNotEmpty(jsonString)) {
                        jsonString = jsonString.substring(1, jsonString.length() - 1);
                    }
                    return jsonString;
                }
                return JsonProcessorsUtils.toJsonString(value);
            }
        },
        BOOLEAN {
            @Override
            public Object convertFrom(Object value) {
                if (String.valueOf(value).matches("^(t|true|yes|y|1)$")) {
                    return true;
                } else if (String.valueOf(value).matches("^(f|false|no|n|0)$")) {
                    return false;
                } else {
                    return null;
                }
            }
        };

        @Override
        public String toString() {
            return this.name().toLowerCase();
        }

        public abstract Object convertFrom(Object value);

        public Object convertFrom(Object value, Object defaultValue) {
            return MoreObjects.firstNonNull(convertFrom(value), defaultValue);
        }
    }
}